// Copyright 2018 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// ------------------------------------------------------------------

#include <TCPChannelResourceBasic.h>

#include <ertps/utils/LocatorAlias.h>
#include <TCPTransportInterface.h>

#include <future>
#include <array>

#include <asio.hpp>

using namespace asio;

namespace vbs {
namespace transport {

using octet = ertps::rtps::octet;
using IPLocator = ertps::rtps::IPLocator;

TCPChannelResourceBasic::TCPChannelResourceBasic(TCPTransportInterface* parent, asio::io_service& service,
                                                 const Locator& locator, uint32_t maxMsgSize)
    : TCPChannelResource(parent, locator, maxMsgSize), service_(service) {}

TCPChannelResourceBasic::TCPChannelResourceBasic(TCPTransportInterface* parent, asio::io_service& service,
                                                 const Locator& locator, const Locator& remote_locator,
                                                 uint32_t maxMsgSize)
    : TCPChannelResource(parent, locator, remote_locator, maxMsgSize), service_(service) {}

TCPChannelResourceBasic::TCPChannelResourceBasic(TCPTransportInterface* parent, asio::io_service& service,
                                                 std::shared_ptr<asio::ip::tcp::socket> socket, const Locator& locator,
                                                 const Locator& remote_locator, uint32_t maxMsgSize)
    : TCPChannelResource(parent, locator, remote_locator, maxMsgSize), service_(service), socket_(std::move(socket)) {}

TCPChannelResourceBasic::~TCPChannelResourceBasic() {}

#if 0
void TCPChannelResourceBasic::connect(const std::shared_ptr<TCPChannelResource>& myself) {
    assert(TCPConnectionType::TCP_CONNECT_TYPE == tcp_connection_type_);
    eConnectionStatus expected = eConnectionStatus::eDisconnected;

    if (connection_status_.compare_exchange_strong(expected, eConnectionStatus::eConnecting)) {
        try {
            ip::tcp::resolver resolver(service_);

            auto endpoints = resolver.resolve(
                {IPLocator::hasWan(locator_) ? IPLocator::toWanstring(locator_) : IPLocator::ip_to_string(locator_),
                 std::to_string(IPLocator::getPhysicalPort(locator_))});

            socket_                                            = std::make_shared<asio::ip::tcp::socket>(service_);
            std::weak_ptr<TCPChannelResource> channel_weak_ptr = myself;

            asio::async_connect(*socket_, endpoints,
                                [this, channel_weak_ptr](std::error_code ec
#if ASIO_VERSION >= 101200
                                                         ,
                                                         ip::tcp::endpoint
#else
                , ip::tcp::resolver::iterator
#endif  // if ASIO_VERSION >= 101200
                                ) {
                                    if (!channel_weak_ptr.expired()) {
                                        parent_->SocketConnected(channel_weak_ptr, ec);
                                    }
                                });
        } catch (const std::system_error& error) {
            logError(RTCP, "Openning socket " << error.what());
        }
    }
}
#else
bool TCPChannelResourceBasic::connect(const std::shared_ptr<TCPChannelResource>& myself) {
    assert(TCPConnectionType::TCP_CONNECT_TYPE == tcp_connection_type_);
    eConnectionStatus expected = eConnectionStatus::eDisconnected;

    if (connection_status_.compare_exchange_strong(expected, eConnectionStatus::eConnecting)) {
        try {
            ip::tcp::resolver resolver(service_);

            // 解析地址和端口
            auto endpoints = resolver.resolve(
                {IPLocator::hasWan(locator_) ? IPLocator::toWanstring(locator_) : IPLocator::ip_to_string(locator_),
                 std::to_string(IPLocator::getPhysicalPort(locator_))});

            // 创建 socket
            socket_ = std::make_shared<asio::ip::tcp::socket>(service_);
            std::weak_ptr<TCPChannelResource> channel_weak_ptr = myself;

            // 同步连接
            asio::connect(*socket_, endpoints);
            //重置local_locator
            endpoint_to_locator(socket_->local_endpoint(), locator_);
            // 通知连接成功
            if (!channel_weak_ptr.expired()) {
                parent_->SocketConnected(channel_weak_ptr, std::error_code());
            }
        } catch (const asio::system_error& error) {
            // 捕获异常并记录日志
            logError(RTCP, "Opening socket failed: " << error.what());

            // 恢复状态为断开连接
            connection_status_.store(eConnectionStatus::eDisconnected);
            return false;
        }
    }
    return true;
}
#endif

void TCPChannelResourceBasic::disconnect() {
    logDebug(RTCP, "TCPChannelResourceBasic disconnect status:" << connection_status() << " alive:" << alive());
    if ((eConnecting < change_status(eConnectionStatus::eDisconnected)) && alive()) {
        logDebug(RTCP, "TCPChannelResourceBasic disconnect");
        auto socket = socket_;

        std::error_code ec;
        (void)socket->shutdown(asio::ip::tcp::socket::shutdown_both, ec);

        service_.post([&, socket_m = std::move(socket)]() {
            try {
                socket_m->cancel();
                socket_m->close();
            } catch (std::exception&) {}
        });
    } else {
        logDebug(RTCP, "not need disconnect status:" << connection_status() << " alive:" << alive());
    }
}

uint32_t TCPChannelResourceBasic::read(octet* buffer, std::size_t size, asio::error_code& ec) {
    std::unique_lock<std::mutex> read_lock(read_mutex_);

    if (eConnecting < connection_status_) {
        return static_cast<uint32_t>(asio::read(*socket_, asio::buffer(buffer, size), transfer_exactly(size), ec));
    }

    return 0;
}

size_t TCPChannelResourceBasic::send(const octet* header, size_t header_size, const octet* data, size_t size,
                                     asio::error_code& ec) {
    size_t bytes_sent = 0;

    if (eConnecting < connection_status_) {
        std::lock_guard<std::mutex> send_guard(send_mutex_);
        if (header_size > 0) {
            std::array<asio::const_buffer, 2> buffers;
            buffers[0] = asio::buffer(header, header_size);
            buffers[1] = asio::buffer(data, size);
            bytes_sent = asio::write(*socket_.get(), buffers, ec);
        } else {
            bytes_sent = asio::write(*socket_.get(), asio::buffer(data, size), ec);
        }
    }

    return bytes_sent;
}

void TCPChannelResourceBasic::async_send(const octet* header, size_t header_size, const octet* data, size_t size,
                                         AsyncSendHandler handler, bool is_default_header) {
    if (eConnecting < connection_status_) {
        std::lock_guard<std::mutex> send_guard(send_mutex_);
        if (header_size > 0) {
            std::array<asio::const_buffer, 2> buffers;
            buffers[0] = asio::buffer(header, header_size);
            buffers[1] = asio::buffer(data, size);
            auto buffers_ptr = std::make_shared<std::array<asio::const_buffer, 2>>(buffers);

            asio::async_write(*socket_, *buffers_ptr,
                              [buffers_ptr, handler_m = std::move(handler), is_default_header, header_size](
                                  const asio::error_code& ec, size_t bytes_transferred) {
                                  logInfo(RTCP, "async_write bytes_transferred: " << bytes_transferred);
                                  if (handler_m) {
                                      if (is_default_header) {
                                          bytes_transferred -= header_size;
                                      }
                                      handler_m(ec, bytes_transferred);
                                  }
                              });
        } else {
            auto buffer_ptr = std::make_shared<asio::const_buffer>(asio::buffer(data, size));

            asio::async_write(*socket_, *buffer_ptr,
                              [buffer_ptr, handler_m = std::move(handler), is_default_header, header_size](
                                  const asio::error_code& ec, size_t bytes_transferred) {
                                  logInfo(RTCP, "async_write bytes_transferred: " << bytes_transferred);
                                  if (handler_m) {
                                      if (is_default_header) {
                                          bytes_transferred -= header_size;
                                      }
                                      handler_m(ec, bytes_transferred);
                                  }
                              });
        }
    } else {
        asio::error_code ec = asio::error::not_connected;
        logError(RTCP, "async_write bytes_transferred: " << 0);
        if (handler) {
            handler(ec, 0);
        }
    }
}

asio::ip::tcp::endpoint TCPChannelResourceBasic::remote_endpoint() const {
    return socket_->remote_endpoint();
}

asio::ip::tcp::endpoint TCPChannelResourceBasic::local_endpoint() const {
    std::error_code ec;
    return socket_->local_endpoint(ec);
}

void TCPChannelResourceBasic::set_options(const TCPTransportDescriptor* options) {
    socket_->set_option(socket_base::receive_buffer_size(options->receiveBufferSize));
    socket_->set_option(socket_base::send_buffer_size(options->sendBufferSize));
    socket_->set_option(ip::tcp::no_delay(options->enable_tcp_nodelay));
}

void TCPChannelResourceBasic::cancel() {
    try {
        socket_->cancel();
    } catch (std::exception&) {}
}

void TCPChannelResourceBasic::close() {
    socket_->close();
}

void TCPChannelResourceBasic::shutdown(asio::socket_base::shutdown_type what) {
    socket_->shutdown(what);
}

}  // namespace transport
}  // namespace vbs
